package worker

import (
	"context"
	"gitee.com/zhengjianfeng/crontab-worker/common"
	"go.etcd.io/etcd/clientv3"
	"go.etcd.io/etcd/mvcc/mvccpb"
	"time"
)

var (
	// 单例
	G_workerJobMgr *WorkerJobMgr
)

// 任务管理器
type WorkerJobMgr struct {
	client  *clientv3.Client
	kv      clientv3.KV
	lease   clientv3.Lease
	watcher clientv3.Watcher
}

// 监听任务变化
func (workerJobMgr *WorkerJobMgr) watchWorkerJob() (err error) {

	// 1、get一下/worker/cron/jobs/目录下所有的任务，并且获取但前集群的revision
	var (
		getResp              *clientv3.GetResponse
		kvPair               *mvccpb.KeyValue
		workerJob            *common.WorkerJob
		workerJobEvent       *common.WorkerJobEvent
		watcherStartRevition int64
		watchChan            clientv3.WatchChan
		watchResp            clientv3.WatchResponse
		watchEvent           *clientv3.Event
		//workerJobName        string
	)

	if getResp, err = workerJobMgr.kv.Get(context.TODO(), common.WORKER_JOB_SAVE_DIR, clientv3.WithPrefix()); err != nil {
		return
	}

	// 但前有哪些任务
	for _, kvPair = range getResp.Kvs {
		// 反序列化json得到Job
		if workerJob, err = common.UnpackWorkerJob(kvPair.Value); err != nil {
			// 忽略解析错误
			err = nil
			continue
		}

		// 构造一个更新Event
		workerJobEvent = common.BuildWorkerJobEvent(common.WORKER_JOB_EVENT_SAVE, workerJob)

		// 把这workerJob同步给scheduler（调度协程）
		G_scheduler.PushWorkerJobEvent(workerJobEvent)

	}

	// 2、从该revition向后监听变化事件
	// 监听协程
	go func() {
		// 从GET时刻的后续版本开始监听变化
		watcherStartRevition = getResp.Header.Revision + 1

		// 启动监听/worker/cron/jobs目录的变化
		watchChan = workerJobMgr.watcher.Watch(context.TODO(), common.WORKER_JOB_SAVE_DIR, clientv3.WithRev(watcherStartRevition), clientv3.WithPrefix())

		// 处理监听
		for watchResp = range watchChan {
			for _, watchEvent = range watchResp.Events {
				switch watchEvent.Type {
				case mvccpb.PUT: // 保存任务
					if workerJob, err = common.UnpackWorkerJob(watchEvent.Kv.Value); err != nil {
						// 忽略错误
						continue
					}

					// 构造一个更新Event
					workerJobEvent = common.BuildWorkerJobEvent(common.WORKER_JOB_EVENT_SAVE, workerJob)
					// 把这workerJob同步给scheduler（调度协程）
					G_scheduler.PushWorkerJobEvent(workerJobEvent)
				}
			}
		}
	}()

	return
}

// 初始化任务管理器
func InitWorkJobMgr() (err error) {

	var (
		config  clientv3.Config
		client  *clientv3.Client
		kv      clientv3.KV
		lease   clientv3.Lease
		watcher clientv3.Watcher
	)

	config = clientv3.Config{
		Endpoints:   G_config.EtcdEndPoints,                                     // 集群地址
		DialTimeout: time.Duration(G_config.EtcdDialTimeout) * time.Millisecond, // 超时时间
	}

	// 建立连接
	if client, err = clientv3.New(config); err != nil {
		return
	}

	// 得到KV和Lease子集
	kv = client.KV
	lease = client.Lease
	watcher = client.Watcher

	// 赋值单例
	G_workerJobMgr = &WorkerJobMgr{
		client:  client,
		kv:      kv,
		lease:   lease,
		watcher: watcher,
	}

	err = G_workerJobMgr.watchWorkerJob()

	return
}

// 创建任务执行锁
func (workerJobMgr *WorkerJobMgr) CreateWorkerJobLock(workerJobName string) (workJobLock *WorkerJoblock) {

	// 返回一把锁
	workJobLock = InitJoblock(workerJobName, workerJobMgr.kv, workerJobMgr.lease)

	return
}
